Language/OS - Multiplatform Resource Library/LANGUAGE OS.iso/lisp/eulisp/feel-075.lha/feel0.75/Libs/CSP/csp.em

Lisp/Scheme | 1992-05-23 | 11KB | 379 lines

;; First bash at CSP type language
;;
;; need 5 constructs:
;; while
;; alt  -- non deterministic selection
;; par  -- concurrent composition
;; seq  -- sequential execution (progn may do)
;; procedures -- creating processes for channels
;; for: equiv to PAR
;; channels -- single-datum things

;; generics
;; c-read, c-write, c-ready
;; connect-processes

(defmodule csp
  (standard0
   semaphores
   loopsII  ;; while
   list-fns) ();; mapvect, collect

   ;; Errors
   (print "loading")
  (defcondition CSP-Error () )
  ;; abstract
  (defstruct Abstract-Channel ()
    ())

  ;; define the generics
  ;; for channels
  (defgeneric c-read (channel))
  (defgeneric c-write (channel data))
  (defgeneric c-ready (channel))

  ;; for processes
  (defgeneric is-csp-process (thread))

  (defgeneric connect-channel-input (channel))
  (defgeneric connect-channel-output (channel))
  ;; should return 'in 'out 'in-out nil

  ;; useful...
  (defun make-communication-sem ()
    (let ((sem (make-semaphore)))
      (open-semaphore sem)
      sem))

  ;; local channels
  (defstruct Channel Abstract-Channel
    ((data-ready initform nil accessor Channel-data-ready)
     (in-sem initform (make-communication-sem)
         accessor Channel-in-sem)
     (out-sem initform (make-communication-sem)
          accessor Channel-out-sem)
     (datum initform '%_Should_not_be_seen_%
        accessor Channel-datum)
     (connected initform nil accessor Channel-connected)
     (input-thread initform nil accessor Channel-input-thread)
     (output-thread initform nil accessor Channel-output-thread))
    constructor make-Channel)

  ;; need to watch for tasks finishing
  (defclass CSP-thread (thread)
    ((parent initform nil accessor CSP-thread-parent))
    metaclass thread-class
    constructor make-CSP-thread)

  (print "defined classes")

  (defmethod initialize-instance ((proto CSP-thread) lst)
    (let ((new-thread (call-next-method)))
      ((setter CSP-thread-parent) new-thread (current-thread))
      new-thread))

  (defmethod c-read ((channel Channel))
    (cond ((not (subthreadp (current-thread)
                (Channel-input-thread channel)))
       (error "Read on wrong end: ~a~%" channel))
      (t
       ((setter Channel-data-ready) channel nil)
       (open-semaphore (Channel-in-sem channel))
       (let ((data (Channel-datum channel)))
         ;; let the other guy out
         ((setter Channel-datum) channel nil)
         (close-semaphore (Channel-out-sem channel))
         (thread-reschedule)
         data))))

  (defmethod c-write ((channel Channel) data)
    (cond ((not (subthreadp (current-thread)
                (Channel-output-thread channel)))
       (error "Write on wrong end: ~a~%" CSP-Error
          'error-value channel)))
    ((setter Channel-datum) channel data)
    (close-semaphore (Channel-in-sem channel))
    ((setter Channel-data-ready) channel data)
    (open-semaphore (Channel-out-sem channel))
    (thread-reschedule))


  (defmethod c-ready ((channel Channel))
    (thread-reschedule)
    (Channel-data-ready channel))

  (defmethod connect-channel-input ((channel Channel))
    (cond ((Channel-input-thread channel)
       (error "Can't reset channel input\n"
          'error-value channel))
      (t ((setter Channel-input-thread) channel
          (current-thread))
         channel)))

  (defmethod connect-channel-output ((channel Channel))
    (cond ((Channel-output-thread channel)
       (error "Can't reset channel output\n" Internal-Error
          'error-value channel))
      (t ((setter Channel-output-thread) channel
          (current-thread))
         channel)))

  (print "and methods")
  ;; channel pairs...
  ;; connections are made with connect-chan-pair
  ;; try u-field first, then l-field

  (defstruct Chan-Pair Abstract-Channel
    ((u-chan initform (make-instance Channel)
         accessor Chan-Pair-u-chan)
     (d-chan initform (make-instance Channel)
         accessor Chan-Pair-d-chan)
     ;; nil 'one 'two
     (connect-count initform nil
            accessor Chan-Pair-connect-count))
    constructor make-Chan-Pair)

  (defconstant *pair-connect-lock* (make-semaphore))

  ;; input, output are compulsory...
  (defstruct Connected-Chan-Pair Abstract-Channel
    ((input initarg input
        accessor Connected-Chan-Pair-input)
     (output initarg output
         accessor Connected-Chan-Pair-output))
    constructor make-Connected-Chan-Pair)
  (print "chans")
  (defmethod initialize-instance ((proto Connected-Chan-Pair) lst)
    (let ((new-obj (call-next-method)))
      (connect-channel-input (Connected-Chan-Pair-input new-obj))
      (connect-channel-output (Connected-Chan-Pair-output new-obj))
      new-obj))

  (defun connect-chan-pair (chan-pair)
    (format t "Connect: count: ~a~%"
        (Chan-Pair-connect-count chan-pair))
    (open-semaphore *pair-connect-lock*)
    (cond
     ((not (Chan-Pair-connect-count chan-pair))
      (let ((new-pair (make-Connected-Chan-Pair
               'input (Chan-Pair-u-chan chan-pair)
               'output (Chan-Pair-d-chan chan-pair))))
    ((setter Chan-Pair-connect-count) chan-pair 'one)
    (close-semaphore *pair-connect-lock*)
    new-pair))
     ((eq (Chan-Pair-connect-count chan-pair) 'one)
      (let ((new-pair (make-Connected-Chan-Pair
               'input (Chan-Pair-d-chan chan-pair)
               'output (Chan-Pair-u-chan chan-pair))))
    ((setter Chan-Pair-connect-count) chan-pair 'two)
    (close-semaphore *pair-connect-lock*)
    new-pair))
     (t (close-semaphore *pair-connect-lock*)
    (error "Tried to connect too often" CSP-Error
           'error-value chan-pair))))

  (print "cp")
  ;; methods...
  (defmethod c-read ((cp Connected-Chan-Pair))
    (c-read (Connected-Chan-Pair-input cp)))

  (defmethod c-ready ((cp Connected-Chan-Pair))
    (prog1 (c-ready (Connected-Chan-Pair-input cp))
      nil))

  (defmethod c-write ((cp Connected-Chan-Pair) data)
    (c-write (Connected-Chan-Pair-output cp) data))

  ;; is thread 1 a subthread of thread 2
  (defun subthreadp (thread1 thread2)
    (cond ((eq thread1 thread2) t)
      ((eq (class-of thread1) thread) nil)
      (t (subthreadp (CSP-thread-parent thread1)
             thread2))))

  (print "channels")
  ;;
  ;; Initializing CSP

  ;; vectors of channels
  (defun make-channel-vector (n)
    (mapvect make-Channel (make-vector n)))

  ;; wait for threads to stop
  (defun await-finish (threads)
    (let ((res (mapcar thread-value threads)))
      res))

  (defun make-ready-csp-thread (fn . args)
    (let ((thread  (make-CSP-thread 'function fn)))
      (apply thread-start (cons thread args))
      thread))

  ;;
  ;; Non-deterministic alternation:
  ;;  given list of pairs of (chan . result)
  ;; return 1st to be true.
  ;; currently busy-wait
  ;; problem: how to make sure of fairness...
  ;; Non blocking wait should do this (I hope)
  (deflocal *weather* 'sunny)

  (defun wait-for-ready-chan (lst)
    (wait-ready-aux (cond ((eq *weather* 'sunny)
               (setq *weather* 'rainy)
               (reverse lst))
              (t (setq *weather* 'sunny)
                 lst))
            nil))

  (defun wait-ready-aux (orig-lst lst)
    (cond ((null lst)
       ;;(thread-reschedule)
       (wait-ready-aux orig-lst orig-lst))
      ((c-ready (caar lst))
       ;;(thread-reschedule)
       (cdar lst))
      (t;;(thread-reschedule)
       (wait-ready-aux orig-lst (cdr lst)))))

  ;;
  ;; macros
  ;;

  ;; PAR foo bar baz => (await-finish (thread-start (lambda () foo))
  ;;                                  (thread-start (lambda () bar)))
  ;; etc

  (defmacro PAR tasks
    `(await-finish (list ,@(mapcar starter tasks))))


  (defun starter (task)
    `(make-ready-csp-thread (lambda ()  ,task)))

  ;; FOR
  ;;
  (defmacro FOR (inits cont-exp increment . body)
    `(let ((@threads@ nil))
       (let (,inits)
     (while ,cont-exp
       (setq @threads@ (cons (make-ready-csp-thread
                  (lambda (,(car inits)) ,@body)
                  ,(car inits))
                 @threads@))
       ,increment))
       (await-finish @threads@)))

  ;; MAPPAR (across a list)
  (defun MAPPAR (fn lst)
    (await-finish (mapcar (lambda (obj)
                (make-ready-csp-thread fn obj))
              lst)))

  ;; SEQ (easy)
  (defmacro SEQ jobs
    `(progn ,@jobs))

  ;; ALT
  ;; (ALT ((in chan-1 x)  (j1 j2 j3))
  ;;      ((guard (in chan-2 y)) (a1 a2 a3)))
  ;;
  ;; get-first-ret should return sym to be executed
  ;;
  ;; (let ((continue (get-first-ret (chan 1)
  ;;                                (if guard chan-2 nil))))
  ;;   (cond ((eq continue g1)
  ;;          (let ((x (c-read chan-1)))
  ;;            j1 j2 j3))
  ;;         ((eq continue g2)
  ;;          (let ((y (c-read chan-2)))
  ;;            a1 a2 a3))
  ;;         (t (error "ALT: unexpected return" CSP-Error))))
  (defmacro ALT alternatives
    (let ((named-alternatives (mapcar (lambda (x) (name-alternative x)) alternatives)))
      `(let ((@continue@ (wait-for-ready-chan
              (collect (lambda (x) x)
                   (list ,@(mapcar make-guard
                           named-alternatives))))))
     (cond ,@(append (mapcar make-alt-stmt named-alternatives)
             '((t (cerror "Unexpected return from alt" clock-tick))))))))

  ;; should be (sym chan var gaurd-expr junk)
  (defun name-alternative (alternative)
    (let ((guard (car alternative))
      (stmt (cdr alternative)))
      (if (eq (car guard) 'IN)
      (list (gensym) (cadr guard) (caddr guard) t stmt)
    (list (gensym) (cadr (reverse guard))
          (caddr (reverse guard))
          (cddr (reverse guard))
          stmt))))

  ;; should be (if (guard) (cons chan sym) nil)
  (defun make-guard (alt)
    `(if ,(cadddr alt) (cons ,(cadr alt) ',(car alt)) nil))

  ;; should be ((eq @continue@ sym) (let ((var continue)) junk))

  (defun make-alt-stmt (alt)
    `((eq @continue@ ',(car alt))
      (let ((,(caddr alt) (c-read ,(cadr alt))))
    ,@(car (last-pair alt)))))

  ;;
  ;; WAIT-FIRST
  ;; like ALT, but taskes list of channels
  ;; (IN-FROM (chan result) lst . cmds)
  (defmacro IN-FROM ( chan-data chans . rest)
    `(let* ((,(car chan-data) (wait-for-ready-chan (mapcar (lambda (x)
                                 (cons x x))
                               ,chans)))
        (,(cadr chan-data) (IN ,(car chan-data))))
       ,@rest))
  ;; in
  ;; (in chan var)

  (defmacro IN (chan . var)
    (cond (var
       `(setq ,(car var) (c-read ,chan))(thread-reschedule))
      (t `(c-read ,chan))))

  ;; out
  ;; (out char value)
  (defmacro OUT (chan value)
    `(progn (c-write ,chan ,value)(thread-reschedule)))

  ;; exports for applications

  (export SEQ IN OUT ALT PAR FOR IN-FROM make-Channel make-Chan-Pair connect-channel-output connect-channel-input
      connect-chan-pair)

  ;; exports cos of macros
  (export await-finish starter make-ready-csp-thread make-alt-stmt make-guard wait-for-ready-chan
      c-write c-read c-ready)
  )


;; Yet another loop macro (untested by me, but did work once).
(defmodule do
  (standard0)
  ()

  (defmacro do (var-init-step-forms end-test-result . body)
    (let ((vars (mapcar car var-init-step-forms))
          (inits (mapcar cadr var-init-step-forms))
          (steps (mapcar caddr var-init-step-forms))
          (end-test (car end-test-result))
          (results (cdr end-test-result)))
    `(let/cc return
       (labels (
         (do-loop ,vars
           (if ,end-test
               (progn ,@results)
               (progn ,@body (do-loop ,@steps)))))
         (do-loop ,@inits)))))

  (export do)

)